#include "spdlog/spdlog.h"
#include <memory>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/asio/redirect_error.hpp>

#include "rtmp_to_flv.hpp"
#include "base/thread/thread_worker.hpp"
#include "core/flv_media_source.hpp"
#include "core/source_manager.hpp"
#include "protocol/flv/flv_tag.hpp"
#include "server/rtmp/rtmp_message/rtmp_message.hpp"
using namespace mms;

RtmpToFlv::RtmpToFlv(ThreadWorker* worker, const std::string& domain_name, const std::string& app_name, const std::string& stream_name) : MediaBridge(domain_name, app_name, stream_name), worker_(worker), check_closable_timer_(worker->get_io_context()) {
    source_ = std::make_shared<FlvMediaSource>(worker, std::weak_ptr<StreamSession>(std::shared_ptr<StreamSession>(nullptr)));
    sink_ = std::make_shared<RtmpMediaSink>(worker);
}

RtmpToFlv::~RtmpToFlv() {
    spdlog::info("destroy RtmpToFlv");
}

bool RtmpToFlv::init(std::weak_ptr<MediaSource> origin_source) {
    origin_source_ = origin_source;
    auto self(shared_from_this());
    boost::asio::co_spawn(worker_->get_io_context(), [this, self]()->boost::asio::awaitable<void> {
        boost::system::error_code ec;
        while (1) {
            check_closable_timer_.expires_from_now(std::chrono::milliseconds(10000));//10s检查一次
            co_await check_closable_timer_.async_wait(boost::asio::redirect_error(boost::asio::use_awaitable, ec));
            if (boost::asio::error::operation_aborted == ec) {
                break;
            }

           // spdlog::info("check flv is_no_sinks_for_time\r\n");
            if (source_->is_no_sinks_for_time(10000)) {//已经10秒没人播放了
                spdlog::info("close RtmpToFlv because no players for 10s");
                close();
                break;
            }
        }
        co_return;
        }, boost::asio::detached);

    auto rtmp_media_sink = std::static_pointer_cast<RtmpMediaSink>(sink_);
    rtmp_media_sink->on_rtmp_message([this, self](const std::vector<std::shared_ptr<RtmpMessage>>& rtmp_msgs)->boost::asio::awaitable<bool> {
        for (auto rtmp_msg : rtmp_msgs) {
            if (rtmp_msg->get_message_type() == RTMP_MESSAGE_TYPE_AUDIO) {
                if (!this->on_audio_packet(rtmp_msg)) {
                    co_return false;
                }
            }
            else if (rtmp_msg->get_message_type() == RTMP_MESSAGE_TYPE_VIDEO) {
                if (!this->on_video_packet(rtmp_msg)) {
                    co_return false;
                }
            }
            else {
                if (!this->on_metadata(rtmp_msg)) {
                    co_return false;
                }
            }
        }

        co_return true;
        });
    return true;
}

bool RtmpToFlv::on_metadata(std::shared_ptr<RtmpMessage> metadata_pkt) {
    std::shared_ptr<FlvTag> script_flv_tag = std::make_shared<FlvTag>();
    script_flv_tag->tag_header.data_size = metadata_pkt->payload_size_;
    script_flv_tag->tag_header.stream_id = metadata_pkt->message_strean_id_;
    script_flv_tag->tag_header.tag_type = FlvTagHeader::ScriptTag;
    script_flv_tag->tag_header.timestamp = metadata_pkt->timestamp_;

    auto script_data = std::make_unique<SCRIPTDATA>();
    script_data->payload = std::string_view((char*)metadata_pkt->payload_, metadata_pkt->payload_size_);
    script_flv_tag->tag_data = std::move(script_data);

    script_flv_tag->raw_data_buf = std::make_unique<uint8_t[]>(11 + metadata_pkt->payload_size_);
    script_flv_tag->raw_data_size = 11 + metadata_pkt->payload_size_;
    script_flv_tag->encode(script_flv_tag->raw_data_buf.get(), script_flv_tag->raw_data_size);
    auto flv_media_source = std::static_pointer_cast<FlvMediaSource>(source_);
    return flv_media_source->on_metadata(script_flv_tag);
}

bool RtmpToFlv::on_audio_packet(std::shared_ptr<RtmpMessage> audio_pkt) {
    std::shared_ptr<FlvTag> audio_flv_tag = std::make_shared<FlvTag>();
    audio_flv_tag->tag_header.data_size = audio_pkt->payload_size_;
    audio_flv_tag->tag_header.stream_id = audio_pkt->message_strean_id_;
    audio_flv_tag->tag_header.tag_type = FlvTagHeader::AudioTag;
    audio_flv_tag->tag_header.timestamp = audio_pkt->timestamp_;

    auto audio_data = std::make_unique<AUDIODATA>();
    audio_data->payload = std::string_view((char*)audio_pkt->payload_, audio_pkt->payload_size_);
    audio_data->decode(audio_pkt->payload_, audio_pkt->payload_size_);
    audio_flv_tag->tag_data = std::move(audio_data);

    audio_flv_tag->raw_data_buf = std::make_unique<uint8_t[]>(11 + audio_pkt->payload_size_);
    audio_flv_tag->raw_data_size = 11 + audio_pkt->payload_size_;
    audio_flv_tag->encode(audio_flv_tag->raw_data_buf.get(), audio_flv_tag->raw_data_size);

    auto flv_media_source = std::static_pointer_cast<FlvMediaSource>(source_);
    return flv_media_source->on_audio_packet(audio_flv_tag);
}

bool RtmpToFlv::on_video_packet(std::shared_ptr<RtmpMessage> video_pkt) {
    std::shared_ptr<FlvTag> video_flv_tag = std::make_shared<FlvTag>();
    video_flv_tag->tag_header.data_size = video_pkt->payload_size_;
    video_flv_tag->tag_header.stream_id = video_pkt->message_strean_id_;
    video_flv_tag->tag_header.tag_type = FlvTagHeader::VideoTag;
    video_flv_tag->tag_header.timestamp = video_pkt->timestamp_;

    auto video_data = std::make_unique<VIDEODATA>();
    video_data->payload = std::string_view((char*)video_pkt->payload_, video_pkt->payload_size_);
    video_data->decode(video_pkt->payload_, video_pkt->payload_size_);
    video_flv_tag->tag_data = std::move(video_data);

    video_flv_tag->raw_data_buf = std::make_unique<uint8_t[]>(11 + video_pkt->payload_size_);
    video_flv_tag->raw_data_size = 11 + video_pkt->payload_size_;
    video_flv_tag->encode(video_flv_tag->raw_data_buf.get(), video_flv_tag->raw_data_size);
    auto flv_media_source = std::static_pointer_cast<FlvMediaSource>(source_);
    return flv_media_source->on_video_packet(video_flv_tag);
}

void RtmpToFlv::close() {
    if (closed_.test_and_set(std::memory_order_acquire)) {
        return;
    }

    check_closable_timer_.cancel();

    if (source_) {
        std::string name = std::string("flv") + source_->get_domain() + source_->get_app() + source_->get_stream();
        SourceManager::get_instance().remove_source(name);
        source_->close();
    }

    auto origin_source = origin_source_.lock();
    if (sink_) {
        auto rtmp_media_sink = std::static_pointer_cast<RtmpMediaSink>(sink_);
        rtmp_media_sink->on_rtmp_message({});
        rtmp_media_sink->close();
    }

    if (origin_source) {
        origin_source->remove_bridge(shared_from_this());
    }
}